package com.shujia.flink.window

import java.lang

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimerService
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo4TopN {
  def main(args: Array[String]): Unit = {

    /**
      * 统计最近10秒种最热蒙的前两个商品
      *
      *
      */
    /*
    itemid,c
001,1
001,1
001,1
002,1
002,1
003,1
003,1
003,1
003,1
001,1
001,1
001,1
002,1
003,1
001,1
002,1
003,1
     */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 6666)

    val eventDS: DataStream[Event] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      Event(split(0), split(1).toInt)
    })

    //1\统计最近10秒商品被点击的数量

    val countDS: DataStream[(String, Int, Long)] = eventDS.keyBy(_.itemId)
      .timeWindow(Time.seconds(10))
      .process(new TimeWindowProcessFunaction)


    //2、统计窗口内最热门的前两个商品
    val topN: DataStream[(String, Int, Long)] = countDS.keyBy(_._3)
      .process(new TopNKeyedProcessFunction)

    topN.print()

    env.execute()

  }


}

case class Event(itemId: String, c: Int)

class TimeWindowProcessFunaction extends ProcessWindowFunction[Event, (String, Int, Long), String, TimeWindow] {


  /**
    * process  每一个窗口中每一个key都会执行一次
    */
  override def process(key: String, context: Context, elements: Iterable[Event], out: Collector[(String, Int, Long)]): Unit = {
    val count: Int = elements.map(_.c).sum


    //获取窗口的结束时间
    val window: TimeWindow = context.window
    val endTIme: Long = window.getEnd


    //发送到下游
    out.collect((key, count, endTIme))
  }
}


class TopNKeyedProcessFunction extends KeyedProcessFunction[Long, (String, Int, Long), (String, Int, Long)] {
  var listState: ListState[(String, Int, Long)] = _

  override def open(parameters: Configuration): Unit = {
    val context: RuntimeContext = getRuntimeContext

    //定义一个集合状态，保存一个窗口中的所有数据
    val listStateDesc: ListStateDescriptor[(String, Int, Long)] =
      new ListStateDescriptor[(String, Int, Long)]("list", createTypeInformation[(String, Int, Long)])

    listState = context.getListState(listStateDesc)
  }

  /**
    * processElement  每一条数据都会执行一次
    *
    */

  override def processElement(value: (String, Int, Long), ctx: KeyedProcessFunction[Long, (String, Int, Long), (String, Int, Long)]#Context, out: Collector[(String, Int, Long)]): Unit = {


    //将当前数据保存到状态中
    listState.add(value)


    //获取时间服务
    val service: TimerService = ctx.timerService()

    //当事件达到窗口的结束时间之后就会执行onTimer
    service.registerProcessingTimeTimer(value._3 + 1000)

  }


  /**
    * onTimer  当定时器到了之后执行
    *
    */

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, (String, Int, Long), (String, Int, Long)]#OnTimerContext, out: Collector[(String, Int, Long)]): Unit = {


    //计算最热门的前两个商品
    val iterable: lang.Iterable[(String, Int, Long)] = listState.get()

    //导入将java集合转换成scala集合的隐式转换
    import scala.collection.JavaConversions._

    val top2Item: List[(String, Int, Long)] = iterable
      .toList
      .sortBy(-_._2) //按数量排降序
      .take(2)


    //将数据发送到下游
    top2Item.foreach(kv => out.collect(kv))


  }
}
